package com.to8to.scene.deamon;

import static com.mongodb.QueryOperators.IN;
import static com.to8to.scene.server.Commons.mongo;
import static com.to8to.thrift.diary.consts.constConstants.TARGET_TYPE_COMMENT;
import static com.to8to.thrift.diary.consts.constConstants.TARGET_TYPE_DIARY;
import static com.to8to.thrift.diary.consts.constConstants.TARGET_TYPE_PRODUCT;
import static com.to8to.thrift.diary.consts.constConstants.TARGET_TYPE_SCENE;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.thrift.TBase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.mongodb.BasicDBObject;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.WriteResult;
import com.mongodb.util.JSON;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.to8to.commons.mq.IMessageHandler;
import com.to8to.commons.mq.RabbitMQConsumer;
import com.to8to.commons.utils.StringUtil;
import com.to8to.commons.utils.ThriftUtil;
import com.to8to.kitt.esb.DefaultDeamon;
import com.to8to.kitt.esb.NodeChangeEvent;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.scene.server.Commons;
import com.to8to.scene.server.SceneDiaryIface;
import com.to8to.scene.server.Tables;
import com.to8to.scene.stat.StatDiaryUpdateEngine;
import com.to8to.scene.stat.StatSceneUpdateEngine;
import com.to8to.scene.stat.StatUpdateIface;
import com.to8to.thrift.diary.advanced.VerifyRequest;
import com.to8to.thrift.diary.comment.CreateCommentRequest;
import com.to8to.thrift.diary.comment.DeleteCommentRequest;
import com.to8to.thrift.diary.consts.constConstants;
import com.to8to.thrift.diary.diary.CreateDiaryRequest;
import com.to8to.thrift.diary.diary.DeleteDiaryRequest;
import com.to8to.thrift.diary.diary.UpdateDiaryRequest;
import com.to8to.thrift.diary.module.TBehavior;
import com.to8to.thrift.diary.module.TComment;
import com.to8to.thrift.diary.module.TDiary;
import com.to8to.thrift.diary.module.TMessage;
import com.to8to.thrift.diary.module.TProduct;
import com.to8to.thrift.diary.module.TScene;
import com.to8to.thrift.diary.product.CreateProductRequest;
import com.to8to.thrift.diary.product.DeleteProductRequest;
import com.to8to.thrift.diary.product.UpdateProductRequest;
import com.to8to.thrift.diary.service.SceneDiaryService;

public class StatWriter extends DefaultDeamon implements IMessageHandler
{
    public static Logger logger = LoggerFactory.getLogger(StatWriter.class);

    public static String LOGBACKConfPath = "crm/common/logback.xml";

    static StatUpdateIface SCENE_UPDATE_ENGINE = new StatSceneUpdateEngine();
    static StatUpdateIface DIARY_UPDATE_ENGINE = new StatDiaryUpdateEngine();

    public static void sendMsg(TBase<?, ?> body)
    {
        String className = body.getClass().getName();

        DBObject req = ThriftUtil.thriftObject2DBObject(body);

        BasicDBObject obj = new BasicDBObject();

        obj.put("className", className);

        obj.put("req", req);

        Commons.producer.send(MessageProperties.PERSISTENT_BASIC, obj.toString().getBytes());
    }

    /**
     * 创建日记（更新各状态下的图片数，各阶段下的日记数）
     * @param req 创建日记的请求信息
     */
    public static void handle(CreateDiaryRequest req)
    {
        // 只是创建日记，先不更新最近更新时间，在审核完成后更新这个时间
        SCENE_UPDATE_ENGINE.refreshDiaryInfo(req.getDiary().getScene_id(), 0, null, false);

        TDiary diary = req.getDiary();
        String user_id = diary.getOwner_id();
        if (diary.getJianli_id() > 0)
        {
            user_id = "" + diary.getJianli_id();
        }
        DIARY_UPDATE_ENGINE.refreshDiaryInfo(diary.get_id(), req.getDiary().getCtime(), user_id, true);
    }

    /**
     * 更新日记（更新各状态下的图片数）
     * @param req 更新日记的请求信息
     */
    public static void handle(UpdateDiaryRequest req)
    {
        // 先不更新最近更新时间，在审核完成后更新这个时间
        SCENE_UPDATE_ENGINE.refreshDiaryInfo(req.getDiary().getScene_id(), 0, null, false);

        TDiary diary = req.getDiary();
        String user_id = diary.getOwner_id();
        if (diary.getJianli_id() > 0)
        {
            user_id = "" + diary.getJianli_id();
        }
        DIARY_UPDATE_ENGINE.refreshDiaryInfo(diary.get_id(), req.getDiary().getCtime(), user_id, true);
    }

    /**
     * 删除日记
     * @param req
     */
    public static void handle(DeleteDiaryRequest req)
    {
        req.getDiary_ids();
        for (String diaryId : req.getDiary_ids())
        {
            refreshDiaryInfo(diaryId, constConstants.ENTITY_STATUS_DEL);
        }
    }

    /**
     * 评论(发送给用户一条消息，告知他被评论了)
     */
    public static void saveMessage(CreateCommentRequest req)
    {
        TMessage message = new TMessage();
        message.setMsg_type(constConstants.MSG_TYPE_COMMENT); // 评论
        message.setCtime(req.getComment().getCtime());

        message.setTgt_type(req.getComment().getTgt_type());

        TDiary diary = null;
        TScene scene = null;
        // 根据评论的目标类型，找到tgt_id
        switch (req.getComment().getTgt_type())
        {
            case constConstants.TARGET_TYPE_SCENE:
                message.setTgt_id(req.getComment().getScene_id());

                scene = getById(Tables.Scene, TScene.class, message.getTgt_id(), Arrays.asList(TScene._Fields.OWNER_ID.getFieldName()));
                if (scene != null)
                {
                    message.setTgt_uid(scene.getOwner_id());
                }
                else
                {
                    return;
                }

                break;
            case constConstants.TARGET_TYPE_DIARY:
                message.setTgt_id(req.getComment().getDiary_id());

                diary =
                    getById(Tables.Diary, TDiary.class, message.getTgt_id(), Arrays.asList(TDiary._Fields.OWNER_ID.getFieldName(), TDiary._Fields.OTIME.getFieldName(), TDiary._Fields.TAG_ID.getFieldName()));
                if (diary != null)
                {
                    message.setTgt_uid(diary.getOwner_id());
                    message.setDiary_otime(diary.getOtime());
                    message.setDiary_tag(diary.getTag_id());
                }
                else
                {
                    return;
                }

                break;
            case constConstants.TARGET_TYPE_COMMENT:
                return;
                //			message.setTgt_id(req.getComment().getParent_comment_id());
                //			
                //			TComment comment = getById(Tables.Comment, TComment.class, message.getTgt_id(), 
                //					Arrays.asList(TComment._Fields.OWNER_ID.getFieldName()));
                //			if (comment != null) {
                //				message.setTgt_uid(comment.getOwner_id());
                //				diary  = getById(Tables.Diary, TDiary.class, comment.getDiary_id(), 
                //						Arrays.asList(TDiary._Fields.OTIME.getFieldName(), 
                //								TDiary._Fields.TAG_ID.getFieldName()));
                //				if (diary != null) {
                //					message.setDiary_otime(diary.getOtime());
                //					message.setDiary_tag(diary.getTag_id());
                //				}
                //			} else {
                //				return;
                //			}
                //			
                //			break;
            case constConstants.TARGET_TYPE_PRODUCT:
                message.setTgt_id(req.getComment().getScene_id());

                scene = getById(Tables.Scene, TScene.class, req.getComment().getScene_id(), Arrays.asList(TScene._Fields.OWNER_ID.getFieldName()));
                if (scene != null)
                {
                    message.setTgt_uid(scene.getOwner_id());
                }
                else
                {
                    return;
                }
        }

        message.setComment_ctime(req.getComment().getCtime());
        message.setComment_content(req.getComment().getContent());
        message.setSrc_id(req.getComment().get_id());
        message.setSrc_uid(req.getComment().getUser_id());

        saveMessage(message);

    }

    /**
     * 创建评论
     * @param req 创建评论的请求
     */
    public static void handle(CreateCommentRequest req)
    {

        TComment comment = req.getComment();
        switch (comment.getTgt_type())
        {
            case TARGET_TYPE_DIARY:
                // 如果在日记下面的回复，要刷新日记、现场的回复数
                DIARY_UPDATE_ENGINE.refreshCommentInfo(comment.getDiary_id(), req.getComment().getCtime(), req.getComment().getUser_id(), true, true);
                SCENE_UPDATE_ENGINE.refreshCommentInfo(comment.getScene_id(), req.getComment().getCtime(), req.getComment().getUser_id(), true, true);
                break;
            case constConstants.TARGET_TYPE_COMMENT:
                DIARY_UPDATE_ENGINE.refreshCommentInfo(comment.getDiary_id(), req.getComment().getCtime(), req.getComment().getUser_id(), true, true);
                SCENE_UPDATE_ENGINE.refreshCommentInfo(comment.getScene_id(), req.getComment().getCtime(), req.getComment().getUser_id(), true, true);
                break;
            case TARGET_TYPE_PRODUCT:
            case TARGET_TYPE_SCENE:
                SCENE_UPDATE_ENGINE.refreshCommentInfo(comment.getScene_id(), req.getComment().getCtime(), req.getComment().getUser_id(), true, true);
                break;
        }

        //回复的回复不产生消息
        if (comment.getTgt_type() != TARGET_TYPE_COMMENT)
            saveMessage(req);
    }

    /**
     * 删除评论
     */
    public static void handle(DeleteCommentRequest req)
    {

        DBObject cond = new BasicDBObject(TComment._Fields._ID.getFieldName(), new BasicDBObject(IN, req.get_ids()));

        // 查找出所有被删除的评论
        DBCursor cursor = mongo.cursor(Tables.Comment, cond, null, null, 0, 0);

        Set<String> scene_ids = new HashSet<String>();
        Set<String> diary_ids = new HashSet<String>();

        while (cursor.hasNext())
        {
            TComment comment = ThriftUtil.dbObject2ThriftObject(cursor.next(), TComment.class);
            // 现场id一定有
            scene_ids.add(comment.getScene_id());
            // 日记id不一定有
            if (!StringUtil.isEmpty(comment.getDiary_id()))
                diary_ids.add(comment.getDiary_id());
        }

        // 刷新评论数（不刷新最近信息）
        for (String diary_id : diary_ids)
            DIARY_UPDATE_ENGINE.refreshCommentInfo(diary_id, 0, null, false, true);

        for (String scene_id : scene_ids)
            SCENE_UPDATE_ENGINE.refreshCommentInfo(scene_id, 0, null, false, true);
    }

    /**
     * 创建商品
     * @param req
     */
    public static void handle(CreateProductRequest req)
    {
        SCENE_UPDATE_ENGINE.refreshProductInfo(req.getProduct().getScene_id(), 0, null, false);
    }

    /**
     * 更新商品请求
     * @param req
     */
    public static void handle(UpdateProductRequest req)
    {
        refreshProductInfo(req.get_id(), constConstants.ENTITY_STATUS_PENDING);
    }

    /**
     * 删除商品
     * @param req
     */
    public static void handle(DeleteProductRequest req)
    {
        for (String productId : req.get_ids())
        {
            refreshProductInfo(productId, constConstants.ENTITY_STATUS_DEL);
            //			DBObject cond = new BasicDBObject("_id", productId);
            //			DBObject fields = new BasicDBObject(TProduct._Fields.SCENE_ID.getFieldName(), 1);
            //			DBObject dbObject = mongo.findOne(Tables.Product, cond, fields, null);
            //			if (dbObject != null) {
            //				Object sceneId = dbObject.get(TProduct._Fields.SCENE_ID.getFieldName());
            //				SCENE_UPDATE_ENGINE.refreshProductInfo(String.valueOf(sceneId));
            //			}
        }

    }

    /**
     * 收藏现场时，保存消息
     * @param req
     */
    @Deprecated
    private static void saveMessage(TBehavior behavior)
    {
        TMessage message = new TMessage();
        message.setMsg_type(constConstants.MSG_TYPE_COLL); // 收藏
        message.setCtime(SceneDiaryIface.getCurrentTimeStamp());

        // 必然是现场id
        message.setTgt_id(behavior.getTgt_id());
        // 必然是现场
        message.setTgt_type(behavior.getTgt_type());

        // 收藏的是现场，所以tgt_type一定是 constConstants.TARGET_TYPE_SCENE
        message.setSrc_uid(behavior.getUser_id());

        TScene scene = getById(Tables.Scene, TScene.class, message.getTgt_id(), Arrays.asList(TScene._Fields.OWNER_ID.getFieldName()));

        if (scene != null)
        {
            message.setTgt_uid(scene.getOwner_id());
            saveMessage(message);
        }
    }

    public static void handle(TBehavior behavior)
    {

        switch (behavior.getTgt_type())
        {
            case constConstants.TARGET_TYPE_SCENE:
            {
                SCENE_UPDATE_ENGINE.syncUpdate(behavior.getTgt_id(), behavior.getBhv_type(), behavior.getStatus());
                break;
            }
            case constConstants.TARGET_TYPE_DIARY:
            {
                DIARY_UPDATE_ENGINE.syncUpdate(behavior.getTgt_id(), behavior.getBhv_type(), behavior.getStatus());
                break;
            }
        }

        // 如果是收藏现场,则需要发送一条消息
        //		if (behavior.getBhv_type() == constConstants.BEHAVIOR_TYPE_COLL
        //				&& behavior.getTgt_type() == constConstants.TARGET_TYPE_SCENE) {
        //			saveMessage(behavior);
        //		}

    }

    /**
     * 审核修改状态
     * @param req
     */
    public static void handle(VerifyRequest req)
    {

        switch (req.getEntity_type())
        {
            case constConstants.TARGET_TYPE_SCENE:
                // 现场的不需要处理
                break;
            case constConstants.TARGET_TYPE_DIARY:
                refreshDiaryInfo(req.getEntity_id(), req.getStatus());
                break;
            case constConstants.TARGET_TYPE_COMMENT:
                // 评论不需要审核，所以在创建时就刷新统计
                //			refreshCommentInfo(req.getEntity_id(), req.getStatus());
                break;
            case constConstants.TARGET_TYPE_PRODUCT:
                // 只有通过审核才更新信息
                refreshProductInfo(req.getEntity_id(), req.getStatus());
                break;
            default:
                break;
        }
    }

    /**
     * 刷新现场的日记信息（各个状态下的日记数）
     * @param diaryId 日记id
     * @param status 审核结果
     */
    private static void refreshDiaryInfo(String diaryId, int status)
    {
        TDiary diary = getById(Tables.Diary, TDiary.class, diaryId, null);
        if (diary == null)
        {
            return;
        }
        if (status == constConstants.ENTITY_STATUS_NORM)
        {
            // 判断该现场最近时间是否大于日记的更新时间
            DBObject cond = new BasicDBObject("_id", diary.getScene_id()).append("stat.recent_time", new BasicDBObject("$gt", diary.getUtime()));
            boolean exist = mongo.exist(Tables.Scene, cond);

            // 如果现场的最近时间小于日记的创建时间，则以日记的创建时间作为现场的最近时间
            if (!exist)
            {
                SCENE_UPDATE_ENGINE.refreshDiaryInfo(diary.getScene_id(), diary.getUtime(), diary.getOwner_id(), true);
                return;
            }
        }
        // 不更新最近信息
        SCENE_UPDATE_ENGINE.refreshDiaryInfo(diary.getScene_id(), 0, null, false);
        SCENE_UPDATE_ENGINE.refreshCommentInfo(diary.getScene_id(), 0, null, false, true);
    }

    private static void refreshProductInfo(String productId, int status)
    {
        TProduct product = getById(Tables.Product, TProduct.class, productId, null);
        if (product == null)
            return;

        /*total price 也相当于一种统计,放在这里比较合适*/
        //        mongo.updateField(Tables.Product, productId, TProduct._Fields.TOTAL_PRICE.getFieldName(), product.getUnit_price() * product.getNum());

        if (status == constConstants.ENTITY_STATUS_NORM)
        {
            // 判断该现场最近时间是否大于商品的更新时间
            DBObject cond = new BasicDBObject("_id", product.getScene_id()).append("stat.recent_time", new BasicDBObject("$gte", product.getUtime()));
            boolean exist = mongo.exist(Tables.Scene, cond);
            // 如果现场的最近时间小于日记的创建时间，则以日记的创建时间作为现场的最近时间
            if (!exist)
            {
                TScene scene = getById(Tables.Scene, TScene.class, product.getScene_id(), Arrays.asList(TScene._Fields.OWNER_ID.getFieldName()));
                if (null != scene)
                    SCENE_UPDATE_ENGINE.refreshProductInfo(product.getScene_id(), product.getUtime(), scene.getOwner_id(), true);
                else
                    logger.error("no such scene exists:{} of the product:{}", product.getScene_id(), product.get_id());
                return;
            }
        }
        SCENE_UPDATE_ENGINE.refreshProductInfo(product.getScene_id(), 0, null, false);
    }

    @SuppressWarnings("unchecked")
    @Override
    public boolean handle(Delivery delivery) throws Exception
    {
        long delivery_tag = delivery.getEnvelope().getDeliveryTag();
        logger.trace("delivery tag:{}", delivery_tag);

        try
        {
            byte[] bytes = delivery.getBody();

            if (bytes == null)
                return true;

            String json = new String(bytes);

            logger.trace("delivery body:{}", json);

            BasicDBObject dbobj = (BasicDBObject) JSON.parse(json);

            if (dbobj == null)
                return true;

            String className = dbobj.getString("className");

            BasicDBObject msg = (BasicDBObject) dbobj.get("req");

            if (msg == null)
                return true;

            Class<?> clazz = Class.forName(className);
            Method method = this.getClass().getMethod("handle", clazz);
            TBase<?, ?> req = ThriftUtil.dbObject2ThriftObject(msg, (Class<? extends TBase<?, ?>>) clazz);

            long start = System.currentTimeMillis();

            // 消息处理方法执行
            method.invoke(null, req);

            logger.debug("handle({} req), 执行耗时：{} 毫秒", clazz.getSimpleName(), System.currentTimeMillis() - start);

        }
        catch (Exception e)
        {
            logger.error("", e);
        }
        return true;//不返回false，否则会死循环
    }

    /**
     * 根据id找对象
     * @param tableName 表名
     * @param clazz 目标对象的class
     * @param _id 对象id
     * @param fields 需要的字段，传null代表所有字段
     * @return 查找的对象
     */
    public static <T extends TBase<?, ?>> T getById(String tableName, Class<T> clazz, String _id, List<String> fields)
    {
        DBObject cond = new BasicDBObject("_id", _id);
        DBObject fs = null;
        if (fields != null)
        {
            fs = new BasicDBObject();
            for (String field : fields)
                fs.put(field, 1);
        }
        DBObject dbObject = mongo.findOne(tableName, cond, fs, null);
        T t = ThriftUtil.dbObject2ThriftObject(dbObject, clazz);
        return t;
    }

    /**
     * 保存消息 
     * @param message
     */
    public static void saveMessage(TMessage message)
    {
        String id = mongo.getAutoincNumber(Tables.Message, true) + "";
        message.set_id(id);

        DBObject dbObject = ThriftUtil.thriftObject2DBObject(message);
        WriteResult wr = mongo.save(Tables.Message, dbObject);
        logger.debug("save message : {}, write result : {}", message, wr);
    }

    @Override
    public void onPrivateConfigChange(Map<String, NodeChangeEvent> config)
    {
        String[] keys = { "common.properties" };

        for (String file_name : keys)
        {
            NodeChangeEvent event = config.get(file_name);

            if (!event.isChange())
                continue;

            byte[] data = event.ndata;

            switch (file_name)
            {
                case "common.properties":
                {
                    Commons.init(data);
                    break;
                }
            }
        }
        super.onPrivateConfigChange(config);
    }

    @Override
    public void onCommonConfigChange(NodeChangeEvent e)
    {
        if (e.path.contains(LOGBACKConfPath))
        {
            String logbackConfig = new String(e.ndata);
            String real = logbackConfig.replace("${java.service}", service).replace("${java.instance}", instance);
            LogUtils.resetConfigByData(real);
        }
    }

    public static void main(String[] args)
    {
        String instance = "stat_writer_2";

        if (args.length > 0)
            instance = args[0];
        StatWriter deamon = new StatWriter();
        deamon.setReportAddressToZK(true);
        deamon.setSyncConfigFromZK(true);
        deamon.setServiceInstance(SceneDiaryService.class.getName(), instance);
        deamon.addCommonConfigPath(LOGBACKConfPath);
        deamon.init();
        logger.info("初始化完成！！！ 启动消息队列进程。。。");
        logger.info("queue:{}", Commons.producer.QueueInfo());
        RabbitMQConsumer con = new RabbitMQConsumer(Commons.producer.getParam(), Commons.queue);
        con.pollBlock(new StatWriter());
    }
}
